
#include "RtBase.h"
#include "RtTransportUdp.h"
#include "RtUdpEndpoint.h"
#include "RtCnRudpBase.h"

#define UDP_SEND_MAX_LEN (16 * 1024)

CRtString RtGetSystemErrorInfo(int inErrno);

//////////////////////////////////////////////////////////////////////////
// class CRtUdpEndpointManager
//////////////////////////////////////////////////////////////////////////

CRtUdpEndpointManager::CRtUdpEndpointManager()
{
	RT_INFO_TRACE_THIS("CRtUdpEndpointManager::CRtUdpEndpointManager");
}

CRtUdpEndpointManager::~CRtUdpEndpointManager()
{
}

RtResult CRtUdpEndpointManager::StartListen(
		IRtAcceptor* aAcceptor, 
		IRtAcceptorConnectorSink *aSink, 
		const CRtInetAddr &aAddrListen,
		BOOL aListenForRudp)
{
	CRtUdpEndpointKey key(aAddrListen);
	UdpEndpointsType::iterator it = m_endpoints.find(key);
	if (it != m_endpoints.end())
		return it->second->StartListen(aAcceptor, aSink, aAddrListen, aListenForRudp);

	CRtUdpEndpoint* pEndpoint = new CRtUdpEndpoint(this);
	RtResult rv = pEndpoint->StartListen(aAcceptor, aSink, aAddrListen, aListenForRudp);
	if (RT_SUCCEEDED(rv))
	{
		UdpEndpointsType::value_type nodeNew(key, pEndpoint);
		m_endpoints.insert(nodeNew);
	}
	else
		delete pEndpoint;

	return rv;
}

RtResult CRtUdpEndpointManager::StopListen(const CRtInetAddr &aAddrListen)
{
	CRtUdpEndpointKey key(aAddrListen);
	UdpEndpointsType::iterator it = m_endpoints.find(key);
	if (it != m_endpoints.end())
		return it->second->StopListen();

	return RT_OK;
}

RtResult CRtUdpEndpointManager::Connect(
		const CRtInetAddr& aPeerAddr, 
		const CRtInetAddr* aLocalAddr,
		CRtTransportUdp*& aTrpt)
{
	if (aLocalAddr)
	{
		CRtUdpEndpointKey key(*aLocalAddr);
		UdpEndpointsType::iterator it = m_endpoints.find(key);
		if (it != m_endpoints.end())
			return it->second->Connect(aPeerAddr, aLocalAddr, aTrpt);
	}

	CRtUdpEndpoint* pEndpoint = new CRtUdpEndpoint(this);
	RtResult rv = pEndpoint->Connect(aPeerAddr, aLocalAddr, aTrpt);
	if (RT_SUCCEEDED(rv))
	{
		UdpEndpointsType::value_type nodeNew(pEndpoint->GetLocalAddr(), pEndpoint);
		std::pair<UdpEndpointsType::iterator, bool> ret = m_endpoints.insert(nodeNew);
		RT_ASSERTE(ret.second==true);
	}
	else {
		delete pEndpoint;
	}

	return rv;
}

//////////////////////////////////////////////////////////////////////////
// class CRtUdpEndpoint
//////////////////////////////////////////////////////////////////////////

CRtUdpEndpoint::CRtUdpEndpoint(CRtUdpEndpointManager* pManager)
: m_pManager(pManager)
, m_pAcceptor(NULL)
, m_pAcceptorSink(NULL)
, m_listenForRudp(FALSE)
{
	RT_INFO_TRACE_THIS("CRtUdpEndpoint::CRtUdpEndpoint");
	CRtThread* pThreadNetwork = CRtThreadManager::Instance()->GetThread(CRtThreadManager::TT_NETWORK);
	RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(pThreadNetwork->GetThreadId()));
	m_pReactor = pThreadNetwork->GetReactor();
}

CRtUdpEndpoint::~CRtUdpEndpoint()
{
	RT_INFO_TRACE_THIS("CRtUdpEndpoint::~CRtUdpEndpoint");
	Close();
}

int CRtUdpEndpoint::OnInput(RT_HANDLE aFd)
{
	static char szBuf[UDP_SEND_MAX_LEN];
	CRtInetAddr addrRecv;
	int nFlag = 0;

	do
	{
		int nRecv = m_socket.RecvFrom(szBuf, sizeof(szBuf), addrRecv, nFlag);
		if (nRecv <= 0) {
			if (nRecv==-1 && errno!=EWOULDBLOCK)
			{
				RT_WARNING_TRACE("CRtUdpEndpoint::OnInput, RecvFrom() failed!"
					" ip=" << addrRecv.GetIpDisplayName() << 
					" port=" << addrRecv.GetPort() <<
					" nRecv=" << nRecv <<
					" err=" << RtGetSystemErrorInfo(errno));

				CRtPairInetAddr addrPairRm(addrRecv, m_localAddr);
				UdpTransportsType::iterator itRm = m_transports.find(addrPairRm);
				if (itRm != m_transports.end()) 
					(*itRm).second->OnClose(aFd, IRtEventHandler::READ_MASK);
			}

			return 0;
		}

		CRtTransportUdp *pTrans = NULL;
		CRtPairInetAddr addrPair(addrRecv, m_localAddr);
		UdpTransportsType::iterator iter = m_transports.find(addrPair);
		if (iter == m_transports.end()) {
			if (m_pAcceptorSink)
			/// The endpoint is in listen state. 
			{
				if (m_listenForRudp && ( nRecv!=2 || szBuf[0]!=(char)RUDP_PDU_SYNC || szBuf[1]!=(char)RUDP_PACKET_PADDING ) )
					/// We expect that the first packet should be rudp sync pdu.
					continue;

				pTrans = new CRtTransportUdp(m_pReactor, this, addrRecv, TRUE);

				/// Don't use operator[] because it does find() and insert().
				//m_Transports[addrPair] = pTrans;
				UdpTransportsType::value_type nodeNew(addrPair, pTrans);
				m_transports.insert(nodeNew);

/*
				RT_INFO_TRACE_THIS("CRtUdpEndpoint::OnInput,"
					" src_ip=" << addrRecv.GetIpDisplayName() << 
					" src_port=" << addrRecv.GetPort() << 
					" dst_ip=" << m_localAddr.GetIpDisplayName() << 
					" dst_port=" << m_localAddr.GetPort()<<
					" num="<<m_transports.size());
*/

				m_pAcceptorSink->OnConnectIndication(RT_OK, pTrans, m_pAcceptor);
			}
		}
		else {
			pTrans = (*iter).second.Get();
		}

		if (pTrans)
			pTrans->OnReceiveCallback(szBuf, nRecv);
	} while(1);

	return 0;
}

int CRtUdpEndpoint::OnClose(RT_HANDLE aFd, MASK aMask)
{
	RT_INFO_TRACE_THIS("CRtUdpEndpoint::OnClose, aFd=" << aFd << " aMask=" << aMask);

	UdpTransportsType::iterator itRm = m_transports.begin();
	while (itRm != m_transports.end()) 
	{
		UdpTransportsType::iterator itTmp = itRm++;
		(*itTmp).second->OnClose(aFd, aMask);
	}
	return 0;
}

RtResult CRtUdpEndpoint::StartListen(
		IRtAcceptor* aAcceptor, 
		IRtAcceptorConnectorSink *aSink, 
		const CRtInetAddr &aAddrListen,
		BOOL aListenForRudp)
{
	if (m_pAcceptor)
	{
		RT_WARNING_TRACE_THIS("CRtUdpEndpoint::StartListen, already listen!");
		return RT_ERROR_ALREADY_INITIALIZED;
	}

	RtResult rv = RT_OK;

	if (m_socket.GetHandle()==RT_INVALID_HANDLE)
	{
		rv = Open(aAddrListen);
		if (RT_FAILED(rv))
		{
//			RT_WARNING_TRACE_THIS("CRtUdpEndpoint::StartListen, Open failed! rv="<<rv);
			return rv;
		}
	}

	m_pAcceptor = aAcceptor;
	m_pAcceptorSink = aSink;
	m_listenForRudp = aListenForRudp;

	RT_INFO_TRACE_THIS("CRtUdpEndpoint::StartListen,"
		" ip=" << aAddrListen.GetIpDisplayName() <<
		" port=" << aAddrListen.GetPort() << 
		" sink=" << aSink << 
		" fd=" << m_socket.GetHandle());

	return RT_OK;
}

RtResult CRtUdpEndpoint::StopListen()
{
	RT_INFO_TRACE_THIS("CRtUdpEndpoint::StopListen, m_pAcceptor="<<m_pAcceptor);

	if (m_pAcceptor==NULL)
		return RT_OK;

	UdpTransportsType::iterator it = m_transports.begin();
	while (it != m_transports.end()) {
		UdpTransportsType::iterator itTmp = it++;
		if (itTmp->second->IsAcceptor())
		{
			itTmp->second->Disconnect(RT_OK);
//			m_transports.erase(itTmp);
		}
	}
	
	m_pAcceptor = NULL;
	m_pAcceptorSink = NULL;

	if (m_transports.empty()) {
		RT_INFO_TRACE_THIS("CRtUdpEndpoint::StopListen, trpt empty.");
		m_pManager->RemoveEndpoint(this);
	}

	return RT_OK;
}

RtResult CRtUdpEndpoint::Connect(
		const CRtInetAddr& aPeerAddr, 
		const CRtInetAddr* aLocalAddr,
		CRtTransportUdp*& aTrpt)
{
	RtResult rv = RT_OK;

	if (m_socket.GetHandle()==RT_INVALID_HANDLE)
	{
		rv = Open(aLocalAddr ? *aLocalAddr : CRtInetAddr(), &aPeerAddr);
		if (RT_FAILED(rv))
		{
//			RT_WARNING_TRACE_THIS("CRtUdpEndpoint::Connect, Open failed! rv="<<rv);
			return rv;
		}
	}

	CRtPairInetAddr pairAddr(aPeerAddr, m_localAddr);
	UdpTransportsType::iterator it = m_transports.find(pairAddr);
	if (it!=m_transports.end()) {
		RT_WARNING_TRACE_THIS("CRtUdpEndpoint::Connect, transport found!");
		errno = RT_ERROR_FOUND;
		return RT_ERROR_FOUND;
	}

	CRtTransportUdp* pTransport = new CRtTransportUdp(m_pReactor, this, aPeerAddr, FALSE);
	UdpTransportsType::value_type nodeNew(pairAddr, pTransport);
	m_transports.insert(nodeNew);
	aTrpt = pTransport;

	RT_INFO_TRACE_THIS("CRtUdpEndpoint::Connect, connect() successful."
		" localip=" << m_localAddr.GetIpDisplayName() <<
		" port=" << m_localAddr.GetPort() <<
		" peerip=" << aPeerAddr.GetIpDisplayName() <<
		" port=" << aPeerAddr.GetPort() <<
		" fd=" << m_socket.GetHandle());

	return RT_OK;
}

void CRtUdpEndpoint::RemoveTransport(const CRtInetAddr &aPeerAddr)
{
	CRtPairInetAddr pairAddr(aPeerAddr, m_localAddr);
	m_transports.erase(pairAddr);

	if (m_pAcceptor==NULL && m_transports.empty())
	{
		RT_INFO_TRACE_THIS("CRtUdpEndpoint::RemoveTransport, trpt & acceptor empty.");
		Close();
		m_pManager->RemoveEndpoint(this);
	}
}

RtResult CRtUdpEndpoint::Open(const CRtInetAddr& aLocalAddr, const CRtInetAddr* aRemoteAddr)
{
	RtResult rv = RT_OK;

	{
		m_localAddr = aLocalAddr;
#if 0
		if (aLocalAddr.GetIpAddrIn4Bytes()==INADDR_ANY && aRemoteAddr)
		{
			m_localAddr.SetIpAddrBy4Bytes(::RtGetLocalIP());
		}
#endif 

		int nRet = m_socket.Open(m_localAddr);
		if (nRet == -1) {
			RT_ERROR_TRACE_THIS("CRtUdpEndpoint::Open, socket Open failed!"
				" addr=" << m_localAddr.GetIpDisplayName() << 
				" port=" << m_localAddr.GetPort() <<
				" err=" << RtGetSystemErrorInfo(errno));
			rv = RT_ERROR_NETWORK_SOCKET_ERROR;
			goto open_fail;
		}

		nRet = m_socket.GetLocalAddr(m_localAddr);
		if (nRet == -1) {
			RT_ERROR_TRACE_THIS("CRtUdpEndpoint::Open, GetLocalAddr failed! err=" << RtGetSystemErrorInfo(errno));
			rv = RT_ERROR_NETWORK_SOCKET_ERROR;
			goto open_fail;
		}

/*
		if (aLocalAddr.GetIpAddrIn4Bytes()==INADDR_ANY && aRemoteAddr)
		{
			// This code block is to get local IP address bound.

			nRet = ::connect(
				(RT_SOCKET)m_socket.GetHandle(), 
				reinterpret_cast<const struct sockaddr *>(aRemoteAddr->GetPtr()), 
				aRemoteAddr->GetSize());
			if (nRet==-1)
			{
#ifdef RT_WIN32
				errno = ::WSAGetLastError();
#endif
				RT_ERROR_TRACE_THIS("CRtUdpEndpoint::Open, socket connect failed! err=" << RtGetSystemErrorInfo(errno));
				rv = RT_ERROR_NETWORK_SOCKET_ERROR;
				goto open_fail;
			}

			nRet = m_socket.GetLocalAddr(m_localAddr);
			if (nRet == -1) {
				RT_ERROR_TRACE_THIS("CRtUdpEndpoint::Open, GetLocalAddr failed! err=" << RtGetSystemErrorInfo(errno));
				rv = RT_ERROR_NETWORK_SOCKET_ERROR;
				goto open_fail;
			}

#ifdef RT_WIN32
			/// Support multi-connections per port under windows platform. 
			CRtInetAddr addrNull;
			const_cast<sockaddr_in*>(addrNull.GetPtr())->sin_family = AF_UNSPEC;
			::connect(
				(RT_SOCKET)m_socket.GetHandle(), 
				reinterpret_cast<const struct sockaddr *>(addrNull.GetPtr()), 
				addrNull.GetSize());
#endif
		}
		else
		{
			nRet = m_socket.GetLocalAddr(m_localAddr);
			if (nRet == -1) {
				RT_ERROR_TRACE_THIS("CRtUdpEndpoint::Open, GetLocalAddr failed! err=" << RtGetSystemErrorInfo(errno));
				rv = RT_ERROR_NETWORK_SOCKET_ERROR;
				goto open_fail;
			}
		}
*/

#ifdef RT_SUPPORT_QOS
		CRtTransportBase::SetQos2Socket(m_socket.GetHandle());
#endif // RT_SUPPORT_QOS

		m_socket.Enable(RT_IPC_SAP::NON_BLOCK);

		DWORD dwRcv = 262144, dwSnd = 262144;
/*
		if (aRemoteAddr==NULL)
			dwRcv = dwSnd = 1048576; // Server side buffer size: 1MB
*/
		int nOption = m_socket.SetOption(SOL_SOCKET, SO_SNDBUF,  &dwSnd, sizeof(DWORD));
		RT_ASSERTE(nOption == 0);
		nOption = m_socket.SetOption(SOL_SOCKET, SO_RCVBUF,  &dwRcv, sizeof(DWORD));
		RT_ASSERTE(nOption == 0);

		rv = m_pReactor->RegisterHandler(this, IRtEventHandler::READ_MASK);
		if (RT_FAILED(rv)) {
			RT_WARNING_TRACE_THIS("CRtUdpEndpoint::Open, RegisterHandler failed! rv=" << rv);
			goto open_fail;
		}
	}

	RT_INFO_TRACE_THIS("CRtUdpEndpoint::Open,"
		" ip=" << m_localAddr.GetIpDisplayName() << 
		" port=" << m_localAddr.GetPort() <<
		" fd=" << m_socket.GetHandle());

	return RT_OK;

open_fail:
	if (m_socket.GetHandle() != RT_INVALID_HANDLE)
	{
		m_pReactor->RemoveHandler(this);
		m_socket.Close();
	}
	return rv;
}

void CRtUdpEndpoint::Close()
{
	m_pAcceptor = NULL;
	m_pAcceptorSink = NULL;
	
	if (m_socket.GetHandle() != RT_INVALID_HANDLE)
	{
		m_pReactor->RemoveHandler(this);
		m_socket.Close();
	}
	
	m_transports.clear();
}

